package com.ilxqx.framework.amqp.service.impl;

import com.ilxqx.framework.amqp.po.SysRabbitMessage;
import com.ilxqx.framework.amqp.po.SysTask;
import com.ilxqx.framework.amqp.service.MessageReceiverService;
import com.ilxqx.framework.util.InfoUtils;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Service;

import java.util.Map;

/**
 * 消息消费服务
 * @author venus
 */
@Service
@Slf4j
public class MessageReceiveServiceImpl implements MessageReceiverService {

    /**
     * 消息处理
     *
     * @param message 消息
     * @param channel 消息通道
     * @param headers 头部信息
     */
    @Override
    @RabbitHandler
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = "${spring.rabbitmq.template.default-receive-queue}", durable = "true"),
            exchange = @Exchange(value = "${spring.rabbitmq.template.exchange}", durable = "true", type = "topic", ignoreDeclarationExceptions = "true"),
            key = "${spring.rabbitmq.template.routing-key}"
    ))
    public void onMessage(@Payload SysRabbitMessage message, Channel channel, @Headers Map<String, Object> headers) {
        log.info("收到新的消息[{}]，开始处理", InfoUtils.stringify(message));
        Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
        // 处理消息
        if (message instanceof SysTask) {
            // 调用处理函数
            try {
                ((SysTask) message).perform();
                // 执行无异常
                channel.basicAck(deliveryTag, false);
            } catch (Exception e) {
                // 执行无异常进行
                log.warn("任务[{}]执行出现异常，任务执行失败：{}", message, e);
            }
        }
    }

}
